package com.intct.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;

public class work2 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟 JSON 数据源
        List<String> jsonData = Arrays.asList(
                "{\"car\":\"鲁B-0001\",\"speed\":\"59\",\"time\":\"2025-08-13 16:30:36\"}",
                "{\"car\":\"鲁B-0002\",\"speed\":\"56\",\"time\":\"2025-08-13 16:26:36\"}",
                "{\"car\":\"鲁B-0003\",\"speed\":\"61\",\"time\":\"2025-08-13 16:36:36\"}",
                "{\"car\":\"鲁B-0004\",\"speed\":\"60\",\"time\":\"2025-08-13 16:35:36\"}",
                "{\"car\":\"鲁B-0005\",\"speed\":\"59\",\"time\":\"2025-08-13 16:26:37\"}",
                "{\"car\":\"鲁B-0001\",\"speed\":\"80\",\"time\":\"2025-08-13 16:26:36\"}",
                "{\"car\":\"鲁B-0002\",\"speed\":\"59\",\"time\":\"2025-08-13 16:26:44\"}",
                "{\"car\":\"鲁B-0003\",\"speed\":\"59\",\"time\":\"2025-08-13 16:26:56\"}",
                "{\"car\":\"鲁B-0003\",\"speed\":\"59\",\"time\":\"2025-08-13 16:28:36\"}",
                "{\"car\":\"鲁B-0003\",\"speed\":\"59\",\"time\":\"2025-08-13 16:26:36\"}"
        );

        DataStream<String> source = env.fromCollection(jsonData);

        ObjectMapper mapper = new ObjectMapper();

        // 解析 JSON 并提取时间戳（用 SimpleDateFormat 代替 DateTimeFormatter）
        DataStream<JsonNode> withWatermarks = source
                .map(json -> mapper.readTree(json))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JsonNode>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<JsonNode>() {
                                            private final SimpleDateFormat sdf =
                                                    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                                            @Override
                                            public long extractTimestamp(JsonNode node, long recordTimestamp) {
                                                try {
                                                    return sdf.parse(node.get("time").asText()).getTime();
                                                } catch (Exception e) {
                                                    return System.currentTimeMillis();
                                                }
                                            }
                                        }
                                )
                );

        // 过滤超速（> 60），统计 5 分钟窗口超速数量
        withWatermarks
                .filter(node -> node.get("speed").asInt() > 60)
                .map(node -> Tuple2.of("overspeed", 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t -> t.f0)
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .reduce((v1, v2) -> Tuple2.of("overspeed", v1.f1 + v2.f1))
                .print("超速车辆数量");

        // 输出超速明细
        withWatermarks
                .filter(node -> node.get("speed").asInt() > 60)
                .map(json -> json.asText())
                .returns(Types.STRING)
                .windowAll(TumblingEventTimeWindows.of(Time.minutes(5)))
                .apply((TimeWindow window, Iterable<String> values, Collector<String> out) -> {
//                    out.collect("窗口: " +
//                            new Timestamp(window.getStart()) + " ~ " +
//                            new Timestamp(window.getEnd()));

                    for (String node : values) {
//                        out.collect("超速明细 -> 车牌: " + node.get("car").asText()
//                                + " 速度: " + node.get("speed").asInt()
//                                + " 时间: " + node.get("time").asText());
                        out.collect("");
                    }
                }).returns(Types.STRING)
                .print("超速车辆明细");

        env.execute("OverSpeedJson");
    }
}